// @Description
// @Author  xiaoyang
// @Date  2022/3/4 5:14 下午

package golang_rpc

import (
	"encoding/json"
	"errors"
	"fmt"
	"golang-rpc/codec"
	"io"
	"log"
	"net"
	"sync"
)

type Call struct {
	Seq           uint64
	ServiceMethod string      // format "<service>.<method>"
	Args          interface{} // arguments to the function
	Reply         interface{} // reply from the function
	Error         error       // if error occurs, it will be set
	Done          chan *Call  // Strobes when call is complete.
}

func (call *Call) done() {
	call.Done <- call
}

// Client represents an RPC Client.
// There may be multiple outstanding Calls associated
// with a single Client, and a Client may be used by
// multiple goroutines simultaneously.
type Client struct {
	cc       codec.Codec      // 消息的编码，解码器
	opt      *Option          // 连接头部（编码方式，机器编号）
	sending  sync.Mutex       // 互斥锁 （保证一个连接不会并发推送多个请求）
	header   codec.Header     //
	mu       sync.Mutex       // protect following
	seq      uint64           // 请求编号 （每个请求编号唯一）
	pending  map[uint64]*Call // 存储未处理完成的请求 （一个连接可以同时监听多个请求）
	closing  bool
	shutdown bool
}

var _ io.Closer = (*Client)(nil)

var ErrShutdown = errors.New("connection is shut down")

// Close
/**
 *  @Author yang
 *  @Description 关闭连接
**/
func (c *Client) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closing {
		return ErrShutdown
	}
	c.closing = true
	return c.cc.Close()
}

// IsAvailable return true if the client does work
/**
 *  @Author yang
 *  @Description 客户端是否运作中
**/
func (c *Client) IsAvailable() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return !c.shutdown && !c.closing
}

/**
 *  @Author yang
 *  @Description 将call暂时存入pending；并且更新请求id
**/
func (c *Client) registerCall(call *Call) (uint64, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.shutdown || c.closing {
		return 0, ErrShutdown
	}
	// 当前请求和回调操作放入客户端
	call.Seq = c.seq
	c.pending[call.Seq] = call
	c.seq++
	return call.Seq, nil
}

/**
 *  @Author yang
 *  @Description 根据请求id,从pending中移除call，并返回call
**/
func (c *Client) removeCall(seq uint64) *Call {
	c.mu.Lock()
	defer c.mu.Unlock()
	call := c.pending[seq]
	delete(c.pending, seq)
	return call
}

/**
 *  @Author yang
 *  @Description 服务端和客户端连接时发生错误，将shutdown设置为true；将错误信息通知所有的pending状态的call
**/
func (c *Client) terminateCalls(err error) {
	c.sending.Lock()
	defer c.sending.Unlock()
	c.mu.Lock()
	defer c.mu.Unlock()
	c.shutdown = true
	for _, call := range c.pending {
		call.Error = err
		call.done()
	}
}

/**
 *  @Author yang
 *  @Description 响应处理
**/
func (c *Client) receive() {
	var err error
	for err == nil {
		var h codec.Header
		if err = c.cc.ReadHeader(&h); err != nil {
			break
		}
		// 获取call对象
		call := c.pending[h.Seq]
		switch {
		case call == nil:
			// call对象不存在
			err = c.cc.ReadBody(nil)
		case h.Error != "":
			// call存在，服务端处理错误
			call.Error = fmt.Errorf(h.Error)
			err = c.cc.ReadBody(nil)
			call.done()
		default:
			// call存在，正常处理
			err = c.cc.ReadBody(call.Reply)
			if err != nil {
				call.Error = errors.New("reading body :" + err.Error())
			}
			call.done()
		}
	}
	// 客户端和服务端连接错误（不可逆，关闭客户端）
	c.terminateCalls(err)
}

/**
 *  @Author yang
 *  @Description 发送数据到目标server端
**/
func (c *Client) send(call *Call) {

	// 通过锁确保同一时间，一个连接只能推送一个信息（一个fd同时只接收一个请求）
	c.sending.Lock()
	defer c.sending.Unlock()

	// call对象保存到client的pending
	seq, err := c.registerCall(call)
	if err != nil {
		call.Error = err
		call.done()
		return
	}

	// client数据推送到server端
	c.header.ServiceMethod = call.ServiceMethod
	c.header.Seq = seq
	c.header.Error = ""
	if err := c.cc.Write(&c.header, call.Args); err != nil {
		// 推送失败--pending中移除call--响应请求
		call := c.removeCall(seq)
		if call != nil {
			call.Error = err
			call.done()
		}
	}
}

func (c *Client) Run(serviceMethod string, args, reply interface{}, done chan *Call) *Call {
	if done == nil {
		done = make(chan *Call, 10)
	} else if cap(done) == 0 {
		log.Panic("golang-rpc client error : done channel is unbuffered")
	}

	call := &Call{
		ServiceMethod: serviceMethod,
		Args:          args,
		Reply:         reply,
		Done:          done,
	}

	c.send(call)
	return call
}

/**
 *  @Author yang
 *  @Description 调用客户端
**/
func (c *Client) Call(serviceMethod string, args, reply interface{}) error {
	call := <-c.Run(serviceMethod, args, reply, make(chan *Call, 1)).Done
	return call.Error
}

/**	创建client端
 *  @Author yang
 *  @Description
**/
func NewClient(conn net.Conn, opt *Option) (*Client, error) {
	// 发送option到server端 (option使用json编码)
	if err := json.NewEncoder(conn).Encode(opt); err != nil {
		log.Println("golang-rpc client error:" + err.Error())
		return nil, err
	}

	// 获取编码对象--构造函数
	f := codec.NewCodecFuncMap[opt.CodecType]
	if f == nil {
		err := fmt.Errorf("%s codec is not existent", opt.CodecType)
		log.Println("golang-rpc client error:" + err.Error())
		return nil, err
	}

	return newClientCodec(f(conn), opt), nil
}

func newClientCodec(cc codec.Codec, option *Option) *Client {
	client := &Client{
		seq:     1, // 初始化请求ID（处理了多少个请求）
		opt:     option,
		cc:      cc,
		pending: make(map[uint64]*Call), // 创建一个空的请求缓冲区
	}

	go client.receive()
	return client
}

/** 创建RPC连接
 *  @Author yang
 *  @Description
**/
func Dial(network string, addr string, opts ...*Option) (client *Client, err error) {
	opt, err := parseOptions(opts...)
	if err != nil {
		return nil, err
	}

	// 连接server端口
	conn, err := net.Dial(network, addr)
	if err != nil {
		return nil, err
	}

	// 关闭连接
	defer func() {
		if client == nil {
			_ = conn.Close()
		}
	}()

	// 创建client端口
	return NewClient(conn, opt)
}

func parseOptions(opts ...*Option) (*Option, error) {
	// 不传入Option使用默认
	if len(opts) == 0 || opts[0] == nil {
		return DefaultOption, nil
	}

	// 如果传入只能传入一个
	if len(opts) != 1 {
		return nil, errors.New("golang-rpc client error : number of options is more than 1")
	}

	opt := opts[0]
	opt.MagicNumber = DefaultOption.MagicNumber
	if opt.CodecType == "" {
		opt.CodecType = DefaultOption.CodecType
	}
	return opt, nil
}
